package com.df.gateway.magager;

import com.df.base.entity.ClientMsg;
import com.df.base.entity.MessageWrapper;
import com.df.gateway.entity.Client;
import com.df.gateway.utils.DubboUtils;
import io.netty.channel.Channel;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ClientManager {
    public static  AttributeKey<String> tokenKey = AttributeKey.newInstance("token");
    private int threadCount = Runtime.getRuntime().availableProcessors() * 2;
    private AtomicInteger index = new AtomicInteger(1);

    //安全问题？
    private Set<Client> todoClient = new HashSet<>();

    //消息收发线程池
    private ThreadPoolExecutor pools = new ThreadPoolExecutor(threadCount,threadCount
            ,60,TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(), (r)->{
                Thread t =  new Thread(()->{
                    r.run();

                });
                t.setName("MessageSendReceiveThread-"+ index.getAndIncrement());
                return t;
            }
    );
    private enum Singleton{
        Instance;
        public ClientManager clientManager = new ClientManager();
    }
    public static ClientManager Instance(){
        return Singleton.Instance.clientManager;
    }
    //channel2client
    private ConcurrentHashMap<Channel, Client> clients = new ConcurrentHashMap<>();
    //token2channel
    private ConcurrentHashMap<String, Client> token2Client = new ConcurrentHashMap<>();
    //TODO消息队列的消息要按token分发到不同的队列么

    public Client getClient(Channel channel){
        return clients.get(channel);
    }

    public boolean setClient(Client client){
        if(client==null || client.getChannel()==null ){
            return false;
        }
        //检查是否已经存在？可能重复添加么
        clients.put(client.getChannel(),client);
        return true;
    }


    public void removeClient(Channel channel){
        if(channel==null){
            return;
        }
        clients.remove(channel);
    }

    public void bindToken(Client client){
        if(client.getToken()==null || client.getToken().isEmpty()){
            return;
        }
        if(client.getChannel()==null){
            return;
        }
        token2Client.put(client.getToken(),client);
    }
    public void unBindToken(String token){
        token2Client.remove(token);
    }
    public Client getClientByToken(String token){
        return token2Client.get(token);
    }



    public void handleMsg(Client client, MessageWrapper msg){
        try {
            //TODO 按client将消息分组后，会不会出现饥饿现象？如果正在处理的客户端一直发送消息，则线程池最多能处理maxthread个客户端
            synchronized (client){
                client.msgs.put(msg);//超过容量其余消息直接丢弃，连接要关闭么？
                if(!todoClient.contains(client)){
                    todoClient.add(client);
                    pools.submit(()->{
                        for (int i = 0; i < 1; i++) {
                            while (!client.msgs.isEmpty()){
                                MessageWrapper poll = client.msgs.poll();
                                if(poll==null){
                                    break;
                                }
                                ClientMsg clientMsg = (ClientMsg) poll;
                                if(poll.isClient2Server()){//向服务端发送消息
                                    DubboUtils.message(clientMsg.getService(),clientMsg.getVersionStr(),poll.getData());
                                }else {
                                    //TODO 判断channel还存活否
                                    client.getChannel().writeAndFlush(poll.getData());
                                    //向客户端返回消息
                                }
                            }
                            synchronized (client){
                                if(client.msgs.isEmpty()){
                                    //TODO如果保证线程安全呢，比如：执行这里前，其他线程判断了容器中有这个，则不加入，此时移除了，但是队列里有新加入的消息
                                    todoClient.remove(client);
                                    break;
                                }else { //还有消息，则继续处理消息
                                    i = -1;
                                }
                            }
                        }

                    });
                }
            }

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void stopListenerMessage(){
        if(!pools.isShutdown()){//检查下关闭时做了哪些工作
            pools.shutdown();
        }
    }

    public class SendReceiveTask implements Runnable{
        Client client;
        public SendReceiveTask(Client client){
            this.client = client;
        }

        @Override
        public void run() {

        }
    }
}
